package com.hu.flink12.api.sql;

import com.hu.flink12.api.entity.Order;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author: hujianjun
 * @Date: 2021/2/9 21:55
 * @Describe: 将DataStream数据转为Table或View然后使用sql进行统计查询
 */
public class DataStream2TableOrView {
    public static void main(String[] args) throws Exception {
        // TODO 1.获取env和tableEnv
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings envSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, envSettings);

        // TODO 2.source
        DataStream<Order> orderA = env.fromCollection(Arrays.asList(
                new Order("001", 1, 3, null),
                new Order("002", 1, 4, null),
                new Order("003", 3, 2, null)
                )
        );

        DataStream<Order> orderB = env.fromCollection(Arrays.asList(
                new Order("004", 2, 3, null),
                new Order("005", 2, 3, null),
                new Order("006", 4, 1, null)
                )
        );

        // TODO 3.将DataStream数据转为Table或View,然后查询
        Table tableA = tableEnv.fromDataStream(orderA, $("orderId").as("order_id"), $("userId").as("user_id"), $("money"), $("timeStamp").as("time_stamp"));
        tableEnv.createTemporaryView("tableB", orderB, $("orderId").as("order_id"), $("userId").as("user_id"), $("money"), $("timeStamp").as("time_stamp"));

        //查询
        String querySql = "select * from " + tableA + " where money>2 union all select * from tableB where money>1";
        Table tableResult = tableEnv.sqlQuery(querySql);

        //打印schema信息
        tableResult.printSchema();

        //不能针对tableResult进行打印，需要转为DataStream才可以打印
        DataStream<Tuple2<Boolean, Row>> tuple2DataStream = tableEnv.toRetractStream(tableResult, Row.class);

        // TODO 4.sink
        tuple2DataStream.print();

        // TODO 5.执行
        env.execute();
    }
}
